package com.zhu.app.dwd;

import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;

/**
 * DWD层下单事物事实表
 * 基于预处理事实表 提取状态为insert的字段即可
 *
 */
public class DWDTradeOrderDetailApp {

    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4);
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);
        //  Configuration configuration = streamTableEnvironment.getConfig().getConfiguration();
        streamTableEnvironment.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        //checkPoint
       /*
        streamExecutionEnvironment.enableCheckpointing(5 * 60000L, CheckpointingMode.EXACTLY_ONCE); //精确一次

        //状态后端
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
         */
        //todo read kafka table dwd_trade_order_pre_process
        //单独写一个方法
        streamTableEnvironment.executeSql("" +
                "create table dwd_trade_order_pre_process ( " +
                "id string, " +
                "order_id string, " +
                "user_id string, " +
                "order_status string, " +
                "sku_id string, " +
                "sku_name string, " +
                "province_id string, " +
                "activity_id string, " +
                "activity_rule_id string, " +
                "coupon_id string, " +
                "date_id string, " +
                "create_time string, " +
                "operate_date_id string, " +
                "operate_time string, " +
                "source_id string, " +
                "source_type string, " +
                "source_type_name string, " +
                "sku_num string, " +
                "split_original_amount string, " +
                "split_activity_amount string, " +
                "split_coupon_amount string, " +
                "split_total_amount string, " +
                "`type` string, " +
                "`old` map<string,string>, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + ZhuKafkaUtil.getKafkaDDL("dwd_trade_order_pre_process", "dwd_trade_order_detail")
        );
        //tableResult.print();
        //Table table = streamTableEnvironment.sqlQuery("select * from dwd_trade_order_pre_process");
        //streamTableEnvironment.createTemporaryView("result",table);
        //streamTableEnvironment.toDataStream(table).print();

        //todo 过滤下单数据
        Table filteredTable = streamTableEnvironment.sqlQuery(
                "select " +
                        "id, " +
                        "order_id, " +
                        "user_id, " +
                        "sku_id, " +
                        "sku_name, " +
                        "province_id, " +
                        "activity_id, " +
                        "activity_rule_id, " +
                        "coupon_id, " +
                        "date_id, " +
                        "create_time, " +
                        "source_id, " +
                        "source_type source_type_code, " +
                        "source_type_name, " +
                        "sku_num, " +
                        "split_original_amount, " +
                        "split_activity_amount, " +
                        "split_coupon_amount, " +
                        "split_total_amount, " +
                        "row_op_ts " +
                        "from dwd_trade_order_pre_process " +
                        "where `type` = 'insert'"
        );
        streamTableEnvironment.createTemporaryView("filtered_table",filteredTable);

        // TODO 5. 创建 Kafka-Connector 下单明细表
        streamTableEnvironment.executeSql("" +
                "create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "sku_id string,\n" +
                "sku_name string,\n" +
                "province_id string,\n" +
                "activity_id string,\n" +
                "activity_rule_id string,\n" +
                "coupon_id string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "source_id string,\n" +
                "source_type_code string,\n" +
                "source_type_name string,\n" +
                "sku_num string,\n" +
                        "split_original_amount string,\n" +
                        "split_activity_amount string,\n" +
                        "split_coupon_amount string,\n" +
                        "split_total_amount string,\n" +
                        "row_op_ts timestamp_ltz(3)\n" +
                        ")" + ZhuKafkaUtil.getKafkaSinkDDL("dwd_trade_order_detail")
        );

        //todo write to kafka topic dwd_trade_order_detail
        streamTableEnvironment.executeSql("insert into dwd_trade_order_detail " +
                "select * from filtered_table ").print();



        streamExecutionEnvironment.execute();

    }

}
